use std::sync::Arc;

use bytes::{Buf, Bytes};
use xactor::{Actor, Addr, Context};

use crate::actors::router::Router;
use crate::prelude::*;

pub struct NatsSubscribeActor {
    addr: Option<Addr<Self>>,
    subject: String,
    message_register: Arc<MessageRegister>,
}

impl NatsSubscribeActor {
    pub fn new(subject: &str, message_register: Arc<MessageRegister>) -> Self {
        Self {
            addr: None,
            subject: subject.to_string(),
            message_register,
        }
    }
}

#[async_trait::async_trait]
impl Actor for NatsSubscribeActor {
    fn name(&self) -> &str {
        "NatsSubscribeActor"
    }

    async fn started(&mut self, ctx: &mut Context<Self>) -> crate::Result<()> {
        info!("NatsSubscribeActor actor:{} started", ctx.actor_id());
        self.addr = Some(ctx.address());
        let router = Router::from_registry().await?;
        let nc = honeybee_nats::connection();
        let message_register = self.message_register.clone();
        nc.subscribe(&self.subject)?.with_handler(move |msg| {
            if msg.reply.is_none() {
                error!("replay is None");
                return Ok(());
            }
            let mut bytes = Bytes::from(msg.data);
            let msg_id = bytes.get_u16();
            if let Err(err) =
                message_register.process_nats(msg.reply.unwrap(), &router, msg_id, bytes)
            {
                error!("process nats message:{} error:{}", msg_id, err);
            }
            Ok(())
        });
        Ok(())
    }
}
